#include "core.h"
#include "PacketProcessor.h"
#include "xStream.h"
#include <google/protobuf/message.h>

xMSG::xMSG(U16 m_id )
{
	msg_ptr = NULL; dwLen = 0; call_guid = 0;from_net_type = ENetDef::NET_NONE;
	dwLen = sizeof(msghead);
	msg_head.msg_id = m_id;
	msg_ptr = new U8[dwLen];
	CGsStream temp(msg_ptr, 0, dwLen);
	temp << msg_head;
}

xMSG::xMSG(U16 m_id,class ::google::protobuf::Message* pMsg, U64 magic_id)
{
	if (pMsg == NULL)
	{
		new(this)xMSG(m_id);
	}
	else
	{
		msg_head.msg_id = m_id;
		msg_head.magic_id = magic_id;
		U32 para_size = pMsg->ByteSize();
		dwLen = sizeof(msghead) + para_size;
		msg_ptr = new U8[dwLen];
		CGsStream temp(msg_ptr, 0, dwLen);
		temp << msg_head;
		pMsg->SerializeToArray(getParameterData(), para_size);
	}
}

xMSG::xMSG(U8* msg_ptr_, U32 msg_len)
{
	call_guid = 0;
	dwLen = msg_len;
	CGsStream temp(msg_ptr_, msg_len, msg_len);
	temp >> msg_head;
	msg_ptr = NULL;
	if (dwLen)
	{
		msg_ptr = new U8[dwLen];
		memcpy(msg_ptr, msg_ptr_, dwLen);
	}
}

xMSG::xMSG(const xMSG& src)
{
	from_net_type = src.from_net_type;
	call_guid = src.call_guid;
	msg_head = src.msg_head; 
	dwLen = src.dwLen;
	msg_ptr = new U8[dwLen];
	memcpy(msg_ptr, src.msg_ptr, dwLen);
}

void xMSG::AddParameter(class ::google::protobuf::Message* pMsg)
{
	U32 para_size = pMsg->ByteSize();
	U32 nNewSize = para_size + sizeof(msghead);
	U8* pNewData = new U8[nNewSize];
	memcpy(pNewData, msg_ptr, sizeof(msghead));
	delete[] msg_ptr;
	msg_ptr = pNewData;
	pMsg->SerializeToArray(getParameterData(), para_size);
	dwLen = nNewSize;
}

void xMSG::release()
{

}

CGsStream& operator<<(CGsStream& stream, xMSG::msghead& value)
{
	return stream << value.msg_id << value.reserved << value.magic_id;
}

CGsStream& operator>>(CGsStream& stream, xMSG::msghead& value)
{
	return stream >> value.msg_id >> value.reserved >> value.magic_id;
}

//--------------------------------------------------------------------------------------------------------------------
IMP_SINGLETON_CLASS(CPacketProcesser)

CPacketProcesser::CPacketProcesser()
: m_bRun(TRUE)
, m_defaultCB(NULL)
{

}

CPacketProcesser::~CPacketProcesser()
{

}

S32 CPacketProcesser::MsgIn(xMSG& msg)
{
	xMSG* p = new xMSG(msg);
	if (p)
	{
		m_record_set.push(p);
		m_event.notify_one();

		return 0;
	}
	
	return -1;
}

void CPacketProcesser::start()
{
	boost::thread t(&CPacketProcesser::ProcessFunction, this);
}

void CPacketProcesser::RegCB(U16 cmd_id, CMD_CB cb)
{
#if 0
	UnReg(cmd_id);
#endif	
	m_mapCB.insert(std::pair<U16, CMD_CB>(cmd_id, cb));
}

void CPacketProcesser::UnReg(U16 cmd_id)
{
	boost::unordered_map<U16, CMD_CB>::iterator it = m_mapCB.find(cmd_id);
	if ( it != m_mapCB.end() )
	{
		m_mapCB.erase(it);
	}
}

void CPacketProcesser::ProcessFunction()
{
	while(m_bRun)
	{
		if(m_event.timed_wait(100))	{
			while(!m_record_set.empty()){
				xMSG* pMsg = NULL;
				if(m_record_set.get(pMsg)){
					boost::unordered_map<U16, CMD_CB>::iterator it = m_mapCB.find(pMsg->getMsgID());
					if (it != m_mapCB.end()){
						CMD_CB pFunc = it->second;
						pFunc(*pMsg);
					}
					else{
						if (m_defaultCB){
							m_defaultCB(*pMsg);
						}
					}
					delete pMsg;
				}
			}
		}
		else //time out
		{
			;
		}
	}
}